package PrjMQTT

import (
	"dgo/goutils/utils"
	"dgo/goutils/ymqtt"
	"fmt"
	"gitee.com/ymofen/supervalue"
	"strings"
	"sync"
)

type ProjMqttService struct {
	Name         string
	url          string
	user         string
	pass         string
	clientid     string
	rootcafile   string
	x509certfile string
	x509keyfile  string

	mqttService *ymqtt.MqttService

	channels []string
	lock     sync.RWMutex
}

func NewProjMqttService() *ProjMqttService {
	rval := &ProjMqttService{
		mqttService: ymqtt.NewMqttService(),
	}
	return rval
}

func (this *ProjMqttService) ConfigName(name string) *ProjMqttService {
	this.Name = name
	return this
}

func (this *ProjMqttService) SetRecvCb(cb func(topic string, data []byte)) {
	mqttservice := this.mqttService
	if mqttservice != nil {
		mqttservice.SetSubscribeCallBack(cb)
	}
}

func (this *ProjMqttService) PublishMsg(topic, msg string) bool {
	mqttservice := this.mqttService
	if mqttservice != nil {
		return mqttservice.PublishMsg(msg, topic)
	} else {
		return false
	}
}

func (this *ProjMqttService) CheckSubscribleChannel(channelid string) bool {
	this.lock.Lock()
	defer this.lock.Unlock()
	for _, v := range this.channels {
		if strings.EqualFold(v, channelid) {
			return false
		}
	}

	this.channels = append(this.channels, channelid)
	if this.mqttService != nil {
		this.mqttService.Subscribe(channelid)
		utils.Log.Infof("[MQTT]订阅MQTT消息:%s", channelid)
	}
	return true
}

func (this *ProjMqttService) PublicJsonMsg(topic, msg string, typeid int) {
	if this.mqttService != nil {
		str := fmt.Sprintf(`{"type":%d,"msg":"%s"}`, typeid, utils.EscapeJsonStr(msg))
		this.mqttService.PublishMsg(str, topic)
	}
}

func (this *ProjMqttService) ReloadConfAndStart2(conf *wrapper.SuperValue) {
	this.Stop()
	if conf == nil {
		utils.Log.Infof("mqtt配置不存在, mqtt服务未启动!")
		return
	}

	this.clientid = conf.AsString("clientid", utils.RandKeyString(24))
	this.url = conf.AsString("url", "")
	this.user = conf.AsString("user", "")
	this.pass = conf.AsString("pass", "")

	this.rootcafile = conf.StringByPath("tls.rootca-pem-file", "")
	this.x509certfile = conf.StringByPath("tls.X509KeyPair-certfile", "")
	this.x509keyfile = conf.StringByPath("tls.X509KeyPair-keyfile", "")

	this.lock.Lock()
	this.channels = strings.Split(conf.AsString("channel", ""), ";")
	this.lock.Unlock()
	if len(this.url) > 0 {
		this.Start()
	}

}

func (this *ProjMqttService) ReloadConfWithoutSubscribleAndStart(conf *wrapper.SuperValue) bool {
	this.Stop()
	if conf == nil {
		utils.Log.Infof("mqtt配置不存在, mqtt服务未启动!")
		return false
	}

	this.clientid = conf.AsString("clientid", utils.RandKeyString(24))
	this.url = conf.AsString("url", "")
	this.user = conf.AsString("user", "")
	this.pass = conf.AsString("pass", "")
	this.rootcafile = conf.StringByPath("tls.rootca-pem-file", "")
	this.x509certfile = conf.StringByPath("tls.X509KeyPair-certfile", "")
	this.x509keyfile = conf.StringByPath("tls.X509KeyPair-keyfile", "")

	if len(this.url) > 0 {
		this.Start()
		return true
	} else {
		return false
	}

}

func (this *ProjMqttService) Stop() {
	if !this.mqttService.IsStarted() {
		return
	}
	this.mqttService.Stop()
	utils.Log.Infof("[%s]mqtt(%s)服务停止", this.Name, this.url)
}

func (this *ProjMqttService) IsStarted() bool {
	return this.mqttService.IsStarted()
}

func (this *ProjMqttService) Start() {
	this.mqttService.SetConfig(this.url, this.user, this.pass, this.clientid)
	rootcafile := this.rootcafile
	if len(rootcafile) > 0 {
		err := this.mqttService.LoadRootCA(rootcafile)
		if err != nil {
			utils.Log.Warnf("[%s]mqtt(%s)加载TLS-RootCA异常:%s", this.Name, this.url, err.Error())
			return
		}

		x509certfile := this.x509certfile
		x509keyfile := this.x509keyfile
		if len(x509certfile) > 0 {
			err := this.mqttService.LoadX509File(x509certfile, x509keyfile)
			if err != nil {
				utils.Log.Warnf("[%s]mqtt(%s)加载TLS-LoadX509File异常:%s", this.Name, this.url, err.Error())
				return
			}
		}
	}

	this.mqttService.SetOnConnectedEvent(func(sender *ymqtt.MqttService) {
		utils.Log.Infof("[MQTT][%s][%s]mqtt服务连接成功", this.Name, this.url)
		this.lock.RLock()
		defer this.lock.RUnlock()
		if len(this.channels) == 0 {
			utils.Log.Infof("[MQTT][%s]没有指定要订阅的主题", this.Name)
		} else {
			for _, v := range this.channels {
				if len(v) > 0 {
					sender.Subscribe(v)
					utils.Log.Infof("[MQTT][%s]订阅MQTT消息:%s", this.Name, v)
				}
			}
		}

	})

	this.mqttService.SetOnDisconnectedEvent(func(sender *ymqtt.MqttService, err error) {
		if err != nil {
			golog.Debugf("Mqtt服务连接断开:%s", err.Error())
		} else {
			golog.Debugf("Mqtt服务连接断开")
		}
	})
	this.mqttService.Start()
}

var (
	PrjMqtt   *ProjMqttService       = NewProjMqttService().ConfigName("PrjMQTT")
	PrjMqttSC *utils.SubscribeCenter = utils.NewSubscribeCenter()
)

func PrjMqttIsStarted() bool {
	return PrjMqtt.IsStarted()
}

func PublishStrMsg(topic, strmsg string) bool {
	if PrjMqtt == nil {
		return false
	}
	rval := PrjMqtt.PublishMsg(topic, strmsg)
	if !rval {
		utils.Log.InfoTypef("MQTT", "[MQTT]发送(%s)消息:%s失败", topic, strmsg)
	}
	return rval

}

func StartProjMqtt(conf *wrapper.SuperValue) bool {
	if conf == nil {
		return false
	}
	PrjMqtt.ReloadConfAndStart2(conf)
	PrjMqtt.SetRecvCb(onMqttRecv)
	return true
}

func StopProjMqtt() {
	PrjMqtt.Stop()
}

func onMqttRecv(topic string, data []byte) {
	//utils.Log.Infof("接收到信息:" + string(data));
	PrjMqttSC.DispatchData(topic, nil, data, nil)
}
